/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.store.ha;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import org.apache.rocketmq.common.ServiceThread;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.common.RemotingUtil;
import org.apache.rocketmq.store.SelectMappedBufferResult;

public class HAConnection {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
    private final HAService haService;
    /**
     * slave与master之间的网络socket channel
     */
    private final SocketChannel socketChannel;
    /**
     * 只用作日志打印.(不重要)
     */
    private final String clientAddr;
    /**
     * 向slave发送数据的服务
     */
    private WriteSocketService writeSocketService;
    /**
     * 从slave读取数据的服务
     * 主要读取的是slave上报的commitLog的最大物理偏移量  long类型 8字节
     */
    private ReadSocketService readSocketService;

    /**
     * 保存启动之后,slave给master发送的 '第一个' offset物理偏移量数据.
     *
     * @see HAService.HAClient#currentReportedOffset slave发给mater的最大物理偏移量.
     */
    private volatile long slaveRequestOffset = -1;

    /**
     * slave发送给master的ack commitLog物理偏移量
     *
     * {@link ReadSocketService} 这个线程读到slave的最大物理偏移量,保存到这里.
     *
     * @see HAService.HAClient#currentReportedOffset slave发给mater的最大物理偏移量.
     */
    private volatile long slaveAckOffset = -1;

    public HAConnection(final HAService haService, final SocketChannel socketChannel) throws IOException {
        this.haService = haService;
        this.socketChannel = socketChannel;
        this.clientAddr = this.socketChannel.socket().getRemoteSocketAddress().toString();
        this.socketChannel.configureBlocking(false); // 设置为非阻塞
        this.socketChannel.socket().setSoLinger(false, -1);
        this.socketChannel.socket().setTcpNoDelay(true);
        this.socketChannel.socket().setReceiveBufferSize(1024 * 64);
        this.socketChannel.socket().setSendBufferSize(1024 * 64);
        this.writeSocketService = new WriteSocketService(this.socketChannel);
        this.readSocketService = new ReadSocketService(this.socketChannel);
        // 客户端连接数量自增
        this.haService.getConnectionCount().incrementAndGet();
    }

    public void start() {
        this.readSocketService.start();
        this.writeSocketService.start();
    }

    public void shutdown() {
        this.writeSocketService.shutdown(true);
        this.readSocketService.shutdown(true);
        this.close();
    }

    public void close() {
        if (this.socketChannel != null) {
            try {
                this.socketChannel.close();
            } catch (IOException e) {
                HAConnection.log.error("", e);
            }
        }
    }

    public SocketChannel getSocketChannel() {
        return socketChannel;
    }

    /**
     * master读取线程
     *
     * master读取到的数据是:broker slave上报的commitLog物理偏移量 —— {@link HAService.HAClient#currentReportedOffset}
     */
    class ReadSocketService extends ServiceThread {
        /**
         * master读取slave发送long类型偏移量时,用的最大buffer最大长度
         * <pre>
         * 说明: 其实slave给master发送的消息只是一个long类型、8字节的 commitLog物理偏移量 而已.
         *       但是为什么要把byteBufferRead设置成 1024*1024 字节呢?
         * 原因:如果slave给master发送了好几个ackOffset,此时master通过一次读取,把好网络I/O中的多个ackOffset读取到byteBufferRead中,
         *      master只需要关注最后一次的ackOffset,就行了. 所以在读取的时候:
         *      {@code
         *                int pos = this.byteBufferRead.position() - (this.byteBufferRead.position() % 8);
         *                long readOffset = this.byteBufferRead.getLong(pos - 8);
         *      }
         * </pre>
         */
        private static final int READ_MAX_BUFFER_SIZE = 1024 * 1024;
        /**
         * 只关注 {@link SelectionKey#OP_READ} 读操作的事件选择器.
         */
        private final Selector selector;
        /**
         * @see HAConnection#socketChannel
         */
        private final SocketChannel socketChannel;
        private final ByteBuffer byteBufferRead = ByteBuffer.allocate(READ_MAX_BUFFER_SIZE);
        private int processPosition = 0;
        private volatile long lastReadTimestamp = System.currentTimeMillis();

        public ReadSocketService(final SocketChannel socketChannel) throws IOException {
            this.selector = RemotingUtil.openSelector();
            this.socketChannel = socketChannel;
            // 注册对 READ 事件.(只关注SOCKET的READ事件.)
            this.socketChannel.register(this.selector, SelectionKey.OP_READ);
            this.setDaemon(true);
        }

        @Override
        public void run() {
            HAConnection.log.info(this.getServiceName() + " service started");

            while (!this.isStopped()) {
                try {
                    this.selector.select(1000);
                    // broker master处理读取事件
                    // (master读取到的数据是:broker slave上报的commitLog物理偏移量 —— HAClient#currentReportedOffset)
                    boolean ok = this.processReadEvent();
                    if (!ok) {
                        HAConnection.log.error("processReadEvent error");
                        break;
                    }

                    // 超时了.expired
                    long interval = HAConnection.this.haService.getDefaultMessageStore().getSystemClock().now() - this.lastReadTimestamp;
                    if (interval > HAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig().getHaHousekeepingInterval()) {
                        log.warn("ha housekeeping, found this connection[" + HAConnection.this.clientAddr + "] expired, " + interval);
                        break;
                    }
                } catch (Exception e) {
                    HAConnection.log.error(this.getServiceName() + " service has exception.", e);
                    break;
                }
            }

            // 下面开始:   优雅关闭 slave socket channel

            this.makeStop();

            writeSocketService.makeStop();

            haService.removeConnection(HAConnection.this);

            HAConnection.this.haService.getConnectionCount().decrementAndGet();

            SelectionKey sk = this.socketChannel.keyFor(this.selector);
            if (sk != null) {
                sk.cancel();
            }

            try {
                this.selector.close();
                this.socketChannel.close();
            } catch (IOException e) {
                HAConnection.log.error("", e);
            }

            HAConnection.log.info(this.getServiceName() + " service end");
        }

        @Override
        public String getServiceName() {
            return ReadSocketService.class.getSimpleName();
        }

        private boolean processReadEvent() {
            // 3次重试.(如果连续3次没有读取到数据,则跳出此方法.)
            int readSizeZeroTimes = 0;

            if (!this.byteBufferRead.hasRemaining()) {
                this.byteBufferRead.flip();
                this.processPosition = 0;
            }

            while (this.byteBufferRead.hasRemaining()) {
                try {
                    // 读数据:从client socketChannel网络I/O中.
                    int readSize = this.socketChannel.read(this.byteBufferRead);
                    if (readSize > 0) {
                        readSizeZeroTimes = 0;
                        this.lastReadTimestamp = HAConnection.this.haService.getDefaultMessageStore().getSystemClock().now();
                        // 如果此次读取的数据长度小于8,则忽略.等待下一次读取,
                        // 因为可能读取到的数据内容是:
                        if ((this.byteBufferRead.position() - this.processPosition) >= 8) {
                            int pos = this.byteBufferRead.position() - (this.byteBufferRead.position() % 8);
                            // slave给master发送的数据只有一个long类型的commitLog文件物理偏移量.
                            // note:为什么getLong(pos-8)时,要指定index=pos-8?
                            //  详见: HAConnection.ReadSocketService#READ_MAX_BUFFER_SIZE
                            long readOffset = this.byteBufferRead.getLong(pos - 8);
                            this.processPosition = pos;

                            HAConnection.this.slaveAckOffset = readOffset;

                            // 小于0,才设置值.也就是说,从启动以来,slave给master发送的第一个offset数据.
                            if (HAConnection.this.slaveRequestOffset < 0) {
                                HAConnection.this.slaveRequestOffset = readOffset;
                                log.info("slave[" + HAConnection.this.clientAddr + "] request offset " + readOffset);
                            }

                            // note:如果slave已同步最新commitLog数据,则会唤醒master broker处理 SEND_MESSAGE 请求时的等待.
                            //  详见此方法上的注释.
                            HAConnection.this.haService.notifyTransferSome(HAConnection.this.slaveAckOffset);
                        }
                    } else if (readSize == 0) {
                        if (++readSizeZeroTimes >= 3) {
                            break;
                        }
                    } else {
                        log.error("read socket[" + HAConnection.this.clientAddr + "] < 0");
                        return false;
                    }
                } catch (IOException e) {
                    log.error("processReadEvent exception", e);
                    return false;
                }
            }

            return true;
        }
    }

    class WriteSocketService extends ServiceThread {
        /**
         * 只关注 {@link SelectionKey#OP_WRITE} 写操作的事件选择器.
         */
        private final Selector selector;
        /**
         * client socket channel
         * slave的网络连接
         */
        private final SocketChannel socketChannel;

        /**
         * master给slave发送消息的请求头.
         * 分别代表:
         * long类型-本次发送数据,commitlog的起始偏移量
         * int 类型-本次发送数据,数据的长度.最大32kb ,详见:  org.apache.rocketmq.store.config.MessageStoreConfig#haTransferBatchSize
         */
        private final int headerSize = 8 + 4;
        private final ByteBuffer byteBufferHeader = ByteBuffer.allocate(headerSize);
        /**
         * 下一次master给slave发送commitLog数据时,从哪个物理偏移量开始.
         */
        private long nextTransferFromWhere = -1;
        private SelectMappedBufferResult selectMappedBufferResult;
        /**
         * 最后一次的数据发送,是否完毕.
         *
         * 都是用 {@link #transferData()} 方法的返回值给这个变量赋值.
         */
        private boolean lastWriteOver = true;
        private long lastWriteTimestamp = System.currentTimeMillis();

        public WriteSocketService(final SocketChannel socketChannel) throws IOException {
            this.selector = RemotingUtil.openSelector();
            this.socketChannel = socketChannel;
            // 只关注可写事件.
            this.socketChannel.register(this.selector, SelectionKey.OP_WRITE);
            this.setDaemon(true);
        }

        @Override
        public void run() {
            HAConnection.log.info(this.getServiceName() + " service started");

            while (!this.isStopped()) {
                try {
                    // 当有可写事件时,会取消阻塞,返回数据.   (因为这个selector,只关注 OP_WRITE 写操作)
                    this.selector.select(1000);

                    // 1、启动后,如果slave还没有给master发送过数据(只是刚连接上端口,还没发送max物理offset的情况)则等待10毫秒,继续循环(本次执行不处理).
                    if (-1 == HAConnection.this.slaveRequestOffset) {
                        Thread.sleep(10);
                        continue;
                    } // 也就是说,这个线程 只有当收到slave发送的maxOffset之后,才开始执行.

                    // 2、启动之后,slave给master发送数据之后,第一次数据同步时:
                    if (-1 == this.nextTransferFromWhere) {
                        // 2.1、如果slave从0开始同步数据,则计算nextTransferFromWhere (下一次从哪个偏移量传输数据)
                        if (0 == HAConnection.this.slaveRequestOffset) {
                            // master broker上,commitLog最大物理偏移量
                            long masterOffset = HAConnection.this.haService.getDefaultMessageStore().getCommitLog().getMaxOffset();
                            // 每个commitLog文件大小.默认是1G
                            int mappedFileSizeCommitLog = HAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig().getMappedFileSizeCommitLog();
                            /*
                             * 这个公式的说明:
                             * 假设目前master最大偏移量masterOffset=3.5G (说明已经生成了3个完整的MappedFile文件(每个为1G),和一个不完整的MappedFile文件,为0.5G),
                             * 则下一次从3.0G开始数据同步,也就是从第4个MappedFile文件开始数据传输.
                             */
                            masterOffset = masterOffset - (masterOffset % mappedFileSizeCommitLog);
                            if (masterOffset < 0) {
                                masterOffset = 0;
                            }
                            this.nextTransferFromWhere = masterOffset;
                            // todo:问题:如果slave从0开始数据同步,也就是slaveRequestOffset=0,则第一次传输数据量是多少?
                        } else {
                            // 2.2、如果slave不是从0开始数据同步,则下一次就直接从slaveRequestOffset 开始同步.
                            this.nextTransferFromWhere = HAConnection.this.slaveRequestOffset;
                        }

                        log.info("master transfer data from " + this.nextTransferFromWhere + " to slave[" + HAConnection.this.clientAddr
                            + "], and slave request " + HAConnection.this.slaveRequestOffset);
                    }

                    // 3、判断最后一次数据发送是否成功
                    if (this.lastWriteOver) {
                        // 距离上一次给slave发送commitLog数据,经过了多长时间.
                        long interval = HAConnection.this.haService.getDefaultMessageStore().getSystemClock().now() - this.lastWriteTimestamp;
                        // ha主从同步,master给slave发送commitLog数据的时间间隔,默认5秒发送一次.
                        // 如果距离上次发送,已经超过5秒,则再次发送commitLog数据.
                        // 如果距离上次发送,没有超过5秒,则不发送.
                        if (interval > HAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig().getHaSendHeartbeatInterval()) {
                            // Build Header
                            this.byteBufferHeader.position(0);
                            this.byteBufferHeader.limit(headerSize);
                            // 告诉slave,下一次从哪个偏移量开始拉取数据.
                            this.byteBufferHeader.putLong(this.nextTransferFromWhere);
                            this.byteBufferHeader.putInt(0);  // 数据长度为0.  只把 nextTransferFromWhere 发送给slave.不发送请求体.
                            this.byteBufferHeader.flip();

                            // master给slave,发送数据.
                            this.lastWriteOver = this.transferData();
                            // 数据未发送成功,则继续下一次循环.
                            if (!this.lastWriteOver)
                                continue;
                        }
                    } else {
                        // 3.2、上一次数据发送没有发送完毕,则继续发送.
                        this.lastWriteOver = this.transferData();
                        if (!this.lastWriteOver)
                            continue;
                    }

                    // 4、根据物理偏移量,读取commitLog文件.
                    SelectMappedBufferResult selectResult = HAConnection.this.haService.getDefaultMessageStore().getCommitLogData(this.nextTransferFromWhere);
                    if (selectResult != null) {
                        int size = selectResult.getSize();
                        // 每次发送数据,最多发送32kb.
                        if (size > HAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig().getHaTransferBatchSize()) {
                            size = HAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig().getHaTransferBatchSize();
                        }

                        long thisOffset = this.nextTransferFromWhere;
                        this.nextTransferFromWhere += size;

                        selectResult.getByteBuffer().limit(size);
                        this.selectMappedBufferResult = selectResult;

                        // Build Header   构建请求头
                        this.byteBufferHeader.position(0);
                        this.byteBufferHeader.limit(headerSize);
                        this.byteBufferHeader.putLong(thisOffset);   // 本次发送数据的起始偏移量
                        this.byteBufferHeader.putInt(size);          // 本次发送数据的长度  最大 32kb
                        this.byteBufferHeader.flip();

                        this.lastWriteOver = this.transferData();    // 发送数据
                    } else {
                        // commitLog没有最新数据需要发送给slave了,则等待一段时间.
                        //     (说明:slave已经和master保持一致了.master没有最新的commitLog文件需要和slave之间同步了,所以就需要等待)
                        // 当commitLog中有新文件时,会notifyAll这个wait. 详见: CommitLog#submitReplicaRequest
                        HAConnection.this.haService.getWaitNotifyObject().allWaitForRunning(100);
                    }
                } catch (Exception e) {
                    // 可能是clientSocketChannel断开连接,然后就异常了.
                    HAConnection.log.error(this.getServiceName() + " service has exception.", e);
                    break;
                }
            } // while (!this.isStopped())  结束.  (有两种情况:1、线程停止了  2、try-catch发生异常,break.)


            // 下面做一些清扫工作,准备断开clientSocketChannel网络连接.

            HAConnection.this.haService.getWaitNotifyObject().removeFromWaitingThreadTable();

            if (this.selectMappedBufferResult != null) {
                this.selectMappedBufferResult.release();
            }

            this.makeStop();

            readSocketService.makeStop();

            haService.removeConnection(HAConnection.this);// 从客户端连接表中,删除当前这个slave的连接.

            SelectionKey sk = this.socketChannel.keyFor(this.selector);
            if (sk != null) {
                sk.cancel();
            }

            try {
                this.selector.close();
                this.socketChannel.close();
            } catch (IOException e) {
                HAConnection.log.error("", e);
            }

            HAConnection.log.info(this.getServiceName() + " service end");
        }

        /**
         * 传输数据
         */
        private boolean transferData() throws Exception {
            // master给slave发送数据时,重试3次.
            int writeSizeZeroTimes = 0;

            // Write Header   发送请求体
            while (this.byteBufferHeader.hasRemaining()) {
                // 使用客户端socket,发送请求头.长度:8+4
                int writeSize = this.socketChannel.write(this.byteBufferHeader);
                if (writeSize > 0) {
                    writeSizeZeroTimes = 0;
                    this.lastWriteTimestamp = HAConnection.this.haService.getDefaultMessageStore().getSystemClock().now();
                } else if (writeSize == 0) {
                    if (++writeSizeZeroTimes >= 3) {
                        break;
                    }
                } else {
                    throw new Exception("ha master write header error < 0");
                }
            }

            // 如果没有请求体需要发送,并且请求头没有剩余数据,表示发送成功.
            if (null == this.selectMappedBufferResult) {
                return !this.byteBufferHeader.hasRemaining();
            }

            // 重试次数,清零
            writeSizeZeroTimes = 0;

            // Write Body   发送请求体
            if (!this.byteBufferHeader.hasRemaining()) {
                while (this.selectMappedBufferResult.getByteBuffer().hasRemaining()) {
                    // 使用client socket发送请求体
                    int writeSize = this.socketChannel.write(this.selectMappedBufferResult.getByteBuffer());
                    if (writeSize > 0) {
                        writeSizeZeroTimes = 0;
                        this.lastWriteTimestamp = HAConnection.this.haService.getDefaultMessageStore().getSystemClock().now();
                    } else if (writeSize == 0) {
                        if (++writeSizeZeroTimes >= 3) {
                            break;
                        }
                    } else {
                        throw new Exception("ha master write body error < 0");
                    }
                }
            }

            // 数据传输是否成功.
            boolean result = !this.byteBufferHeader.hasRemaining() && !this.selectMappedBufferResult.getByteBuffer().hasRemaining();

            // 需要发送的数据没有剩余,则释放.并设置为null.(为了下一次传输数据做初始化准备.)
            if (!this.selectMappedBufferResult.getByteBuffer().hasRemaining()) {
                this.selectMappedBufferResult.release();
                this.selectMappedBufferResult = null;
            }

            return result;
        }

        @Override
        public String getServiceName() {
            return WriteSocketService.class.getSimpleName();
        }

        @Override
        public void shutdown() {
            super.shutdown();
        }
    }
}
